package com.bw.Dim;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// DimSource.java
public class DimSource {
    public static void main(String[] args) {
        // 设置执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1. 配置MySQL CDC源 - 捕获行为表变更
        MySqlSource<String> behaviorSource = MySqlSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .databaseList("gmall_bd") // 数据库名
                .tableList("gmall_bd.dy_table") // 表名
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema()) // 将变更事件序列化为JSON字符串
                .build();

        // 2. 配置MySQL CDC源 - 捕获类型表变更
        MySqlSource<String> typeSource = MySqlSource.<String>builder()
                .hostname("hadoop101")
                .port(3306)
                .databaseList("gmall_bd")
                .tableList("gmall_bd.type_table")
                .username("root")
                .password("123456")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();

        // 3. 创建数据流
        DataStream<String> behaviorStream = env.fromSource(behaviorSource, WatermarkStrategy.noWatermarks(), "MySQL dy_table Source");
        DataStream<String> typeStream = env.fromSource(typeSource, WatermarkStrategy.noWatermarks(), "MySQL type_table Source");
    }

}